package service

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"github.com/google/uuid"
	"mall/api/inventory"
	"mall/api/mall"
	"mall/api/pay"
	"mall/api/prod"
	"mall/internal/conf"
	"mall/internal/constant"
	"mall/internal/model"
	"mall/pkg/ecode"
	"mall/pkg/log"
	"mall/pkg/nsq"
	"mall/pkg/orm"
	"mall/pkg/request_key"
	"mall/pkg/util"
	"mall/pkg/wechat"
	"time"
)

type msgRetryKey struct{}

//var nsqMarshaler = jsonpb.JSONPb{EnumsAsInts: true, OrigName: true}

// NSQ消息恢复
func (s *Service) NSQRetry(ctx context.Context, req *mall.CommonRetryRequest) (resp *mall.DefaultResp, err error) {
	limit := int(req.Limit)
	if limit == 0 {
		limit = 1000
	}
	var messages, sendMsgs []*model.NsqMessage
	if messages, err = s.dao.FindRetryMessages(ctx, limit); err != nil {
		return
	}
	// 标记上下文为retry版本
	if len(messages) != 0 {
		requestId := request_key.GetRequestID(ctx)
		// 异步执行，不阻塞请求，防止外部请求超时时，context cancel导致后续处理失败
		go func() {
			for _, msg := range messages {
				if _, ok := processorMap[msg.Topic]; ok {
					// 消息接收重试任务
					subCtx := request_key.GenerateContextWithRequestID(requestId)
					subCtx = context.WithValue(subCtx, msgRetryKey{}, msg)
					if msgError := processorMap[msg.Topic](subCtx, msg.Payload); msgError != nil {
						log.FromContext(subCtx).Infow("nsq",
							"status_code", "Fail-Unknown",
							"method", "retry",
							"path", msg.Topic,
							"body", msg.MsgKey,
							"error", msgError.Error(),
							"stacktrace", fmt.Sprintf("msg retry error: %+v", err),
						)
					}
				} else {
					// 消息发送重试任务
					sendMsgs = append(sendMsgs, msg)
				}
			}
			if len(sendMsgs) != 0 {
				// 批量重试
				subCtx := request_key.GenerateContextWithRequestID(requestId)
				err = s.dao.Transaction(subCtx, func(subCtx context.Context) (err error) {
					err = s.dao.RetryNSQMessages(subCtx, sendMsgs)
					return
				})
			}
		}()
	}
	resp = &mall.DefaultResp{Status: true}
	return
}

// 商品库更新
func (s *Service) ProdUpdateHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &prod.ProdUpdateMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("prod_update_notify: %s", string(msgBody)),
		msgBody, &err)
	// 1. 获取在货架商品ID序列
	var shelfProds []*model.ShelfProd
	if shelfProds, err = s.dao.FindShelfProdsByRawProdIds(ctx, req.SgId, req.Ids); err != nil {
		return
	}
	affectProdIdMap := make(util.Int64BoolMap)
	for _, shelfProd := range shelfProds {
		affectProdIdMap[shelfProd.ProdId] = true
	}
	if len(affectProdIdMap) != 0 {
		var shelfSkus []*model.ShelfSku
		if shelfSkus, err = s.dao.FindShelfSkusByRawProdIds(
			ctx, req.SgId, affectProdIdMap.GetKeys()); err != nil {
			return
		}
		skuMap := make(map[uint64]map[string]*model.ShelfSku)
		for _, sku := range shelfSkus {
			if _, ok := skuMap[sku.ProdId]; !ok {
				skuMap[sku.ProdId] = make(map[string]*model.ShelfSku)
			}
			skuMap[sku.ProdId][fmt.Sprintf("%d-%d", sku.SkuId,
				sku.PStoreId)] = sku
		}
		now := time.Now()
		var prodResp *prod.SelectionProdsFullResponse
		if prodResp, err = s.downstream.SelectionProdsByProdIds(
			ctx, req.SgId, req.Ids, true); err != nil {
			return
		}
		prodMap := make(map[uint64]*prod.SelectionProdView)
		customPriceMap := make(map[string]uint64)
		for _, rawProd := range prodResp.Prods {
			prodMap[rawProd.Id] = rawProd
		}
		for _, customPrice := range prodResp.CustomSkus {
			customPriceMap[fmt.Sprintf("%d-%d", customPrice.SkuId,
				customPrice.StoreId)] = uint64(customPrice.Price)
		}

		for _, shelfProd := range shelfProds {
			rawProd := prodMap[shelfProd.ProdId]
			shelfProd.SyncName = rawProd.Name
			shelfProd.SyncTypeId = rawProd.ProdTypeId
			shelfProd.SyncPic = rawProd.Pic
			shelfProd.SyncUnit = rawProd.Unit
			shelfProd.SyncGroupDetails = rawProd.GroupDetail
			shelfProd.SyncGroups = rawProd.GroupList
			shelfProd.SyncTime = now
			if err = s.dao.AutoUpdate(ctx, shelfProd); err != nil {
				return
			}
			for _, sku := range rawProd.Skus {
				var syncPrice uint64
				skuKey := fmt.Sprintf("%d-%d", sku.Id, shelfProd.PStoreId)
				if menuPrice, hasMenuPrice := customPriceMap[skuKey]; hasMenuPrice {
					syncPrice = menuPrice
				} else {
					syncPrice = uint64(sku.Price)
				}
				if shelfSku, ok := skuMap[shelfProd.ProdId][skuKey]; ok {
					// 存在记录
					delete(skuMap[shelfProd.ProdId], skuKey)
					shelfSku.SyncTime = now
					shelfSku.SyncWeight = sku.Weight
					shelfSku.SyncItems = sku.Items
					shelfSku.SyncPrice = syncPrice
					if err = s.dao.AutoUpdate(ctx, shelfSku); err != nil {
						return
					}
				} else {
					// 不存在时需要创建记录
					shelfSku = &model.ShelfSku{
						SkuId:        sku.Id,
						PStoreId:     shelfProd.PStoreId,
						ProdId:       shelfProd.ProdId,
						SgId:         req.SgId,
						CustomPrice:  syncPrice,
						IsCustomized: false,
						SyncItems:    sku.Items,
						SyncPrice:    syncPrice,
						SyncWeight:   sku.Weight,
						SyncTime:     now,
						Operator:     shelfProd.Operator,
					}
					if err = s.dao.Create(ctx, shelfSku); err != nil {
						return
					}
				}
			}
			// 剩余均为已删除规格
			for _, subSkuMap := range skuMap {
				for _, sku := range subSkuMap {
					if err = s.dao.DeleteShelfSku(ctx, sku.SkuId, sku.PStoreId); err != nil {
						return
					}
				}
			}
		}
	}
	return
}

// 商品库删除
func (s *Service) ProdDeleteHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &prod.ProdDeleteMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("prod_delete_notify: %s", string(msgBody)),
		msgBody, &err)
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		if err = s.dao.DeleteShelfProdsByProdIds(ctx, req.Ids, req.SgId); err != nil {
			return
		}
		return
	}); err != nil {
		return
	}
	return
}

// 供货组更新
func (s *Service) MenuUpdateHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &prod.MenuUpdateMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("menu_update_notify: %s", string(msgBody)),
		msgBody, &err)
	syncTime := time.Unix(req.Timestamp, 0).In(util.FixedLocation)
	if req.IsAllowModifyChangedToForbidden && len(req.Stores) != 0 {
		// 编辑菜谱触发改价规则从允许改价 => 不允许改价，更新规格表即可
		affectStoreIdMap := make(util.Int32BoolMap)
		for _, store := range req.Stores {
			affectStoreIdMap[store.StoreId] = true
		}
		if err = s.dao.ResetSkuPrice(ctx, req.SgId, affectStoreIdMap.GetKeys(), syncTime); err != nil {
			return
		}
	} else {
		for _, store := range req.Stores {
			if len(store.DelIds) != 0 {
				if err = s.dao.DeleteShelfProds(ctx, store.StoreId, store.DelIds); err != nil {
					return
				}
			}
			if len(store.UpdateIds) != 0 {
				var shelfProds []*model.ShelfProd
				if shelfProds, err = s.dao.FindOnShelfProdsByProdIds(
					ctx, req.SgId, store.StoreId, store.UpdateIds); err != nil {
					return
				}
				affectProdIdMap := make(util.Int64BoolMap)
				for _, shelfProd := range shelfProds {
					affectProdIdMap[shelfProd.ProdId] = true
				}
				if len(affectProdIdMap) != 0 {
					var (
						shelfSkus    []*model.ShelfSku
						updatedProds []*prod.SelectionProdView
					)
					if shelfSkus, err = s.dao.FindShelfSkusByProdIds(
						ctx, req.SgId, store.StoreId, affectProdIdMap.GetKeys()); err != nil {
						return
					}
					skuMap := make(map[uint64]map[string]*model.ShelfSku)
					for _, sku := range shelfSkus {
						if _, ok := skuMap[sku.ProdId]; !ok {
							skuMap[sku.ProdId] = make(map[string]*model.ShelfSku)
						}
						skuMap[sku.ProdId][fmt.Sprintf("%d-%d", sku.SkuId,
							sku.PStoreId)] = sku
					}
					now := time.Now()
					if updatedProds, err = s.downstream.SelectionMenuProdsByProdIds(
						ctx, req.SgId, store.StoreId, affectProdIdMap.GetKeys()); err != nil {
						return
					}
					prodMap := make(map[uint64]*prod.SelectionProdView)
					for _, rawProd := range updatedProds {
						prodMap[rawProd.Id] = rawProd
					}
					for _, shelfProd := range shelfProds {
						rawProd := prodMap[shelfProd.ProdId]
						shelfProd.SyncName = rawProd.Name
						shelfProd.SyncTypeId = rawProd.ProdTypeId
						shelfProd.SyncPic = rawProd.Pic
						shelfProd.SyncUnit = rawProd.Unit
						shelfProd.SyncGroupDetails = rawProd.GroupDetail
						shelfProd.SyncGroups = rawProd.GroupList
						shelfProd.SyncTime = now
						if err = s.dao.AutoUpdate(ctx, shelfProd); err != nil {
							return
						}
						for _, sku := range rawProd.Skus {
							var syncPrice uint64
							skuKey := fmt.Sprintf("%d-%d", sku.Id, shelfProd.PStoreId)
							syncPrice = uint64(sku.Price)
							if shelfSku, ok := skuMap[shelfProd.ProdId][skuKey]; ok {
								// 存在记录
								delete(skuMap[shelfProd.ProdId], skuKey)
								shelfSku.SyncTime = now
								shelfSku.SyncWeight = sku.Weight
								shelfSku.SyncItems = sku.Items
								shelfSku.SyncPrice = syncPrice
								if err = s.dao.AutoUpdate(ctx, shelfSku); err != nil {
									return
								}
							} else {
								// 不存在时需要创建记录
								shelfSku = &model.ShelfSku{
									SkuId:        sku.Id,
									PStoreId:     shelfProd.PStoreId,
									ProdId:       shelfProd.ProdId,
									SgId:         req.SgId,
									CustomPrice:  syncPrice,
									IsCustomized: false,
									SyncItems:    sku.Items,
									SyncPrice:    syncPrice,
									SyncWeight:   sku.Weight,
									SyncTime:     now,
									Operator:     shelfProd.Operator,
								}
								if err = s.dao.Create(ctx, shelfSku); err != nil {
									return
								}
							}
						}
						// 剩余均为已删除规格
						for _, subSkuMap := range skuMap {
							for _, sku := range subSkuMap {
								if err = s.dao.DeleteShelfSku(ctx, sku.SkuId, sku.PStoreId); err != nil {
									return
								}
							}
						}
					}
				}
			}

		}
	}
	return
}

// 订单定时任务
func (s *Service) InnerOrderScheduleHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &mall.MallOrderScheduleMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer func(rawError *error) {
		topic, _ := ctx.Value(nsq.TopicContextKey).(string)
		if *rawError == nil {
			log.FromContext(ctx).Warnw("nsq",
				"status_code", "Success",
				"method", "consume",
				"path", topic,
				"body", string(msgBody),
				"error", "",
				"stacktrace", "",
			)
		} else {
			log.FromContext(ctx).Warnw("nsq",
				"status_code", "Fail-Unknown",
				"method", "consume",
				"path", topic,
				"body", string(msgBody),
				"error", (*rawError).Error(),
				"stacktrace", fmt.Sprintf("error: %+v", *rawError),
			)
		}
	}(&err)
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		switch req.Type {
		case mall.ScheduleType_SCHEDULE_TYPE_CLOSE:
			// 关闭请求
			if err = s.closeBizRequest(ctx, req.OrderId); err != nil {
				return
			}
		case mall.ScheduleType_SCHEDULE_TYPE_PAYING:
			// 前端已触发支付中，需要查询支付系统结果
			if err = s.queryPayment(ctx, req.OrderId); err != nil {
				return
			}
		case mall.ScheduleType_SCHEDULE_TYPE_PROCESS_CREATE:
			// 创建履约单
			if err = s.createProcessOrder(ctx, req.OrderId); err != nil {
				return
			}
		case mall.ScheduleType_SCHEDULE_TYPE_AUTO_CONFIRM:
			// 创建履约单
			if err = s.autoConfirm(ctx, req.OrderId); err != nil {
				return
			}
		case mall.ScheduleType_SCHEDULE_TYPE_REFUND_CREATE:
			// 创建履约单
			if err = s.createRefund(ctx, req.OrderId); err != nil {
				return
			}
		}
		return
	}); err != nil {
		return
	}
	return
}

// 支付结果通知，必定为中泰
func (s *Service) PayResultHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &pay.PayOrderView{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("pay_notify: %s", string(msgBody)),
		msgBody, &err)
	if req.BizType != pay.BizType_BIZ_MALL {
		return
	}
	var processSchedule *model.Schedule
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		request := &model.OrderRequest{}
		if err = s.dao.LockRequest(ctx, request, req.BizNo); err != nil {
			return
		}
		switch pay.PayOrderStatus(request.PayStatus) {
		case req.Status:
			// 与本地状态一致，不处理
			return
		case pay.PayOrderStatus_PAY_ORDER_STATUS_PAYING:
			if err = s.dao.DeleteSchedule(ctx,
				uint8(mall.ScheduleType_SCHEDULE_TYPE_PAYING), request.ID); err != nil {
				return
			}
			fallthrough
		case pay.PayOrderStatus_PAY_ORDER_STATUS_INIT:
			if processSchedule, err = s.updatePayResultInTransaction(ctx, req, request); err != nil {
				return
			}
			return
		}
		return
	}); err != nil {
		return
	}
	if processSchedule != nil {
		if err = s.createProcessOrder(ctx, processSchedule.OrderId); err != nil {
			return
		}
	}
	return
}

// 退款结果通知
func (s *Service) RefundResultHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &pay.RefundOrderView{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("refund_notify: %s", string(msgBody)),
		msgBody, &err)
	if req.BizType != pay.BizType_BIZ_MALL {
		return
	}
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		apply := &model.Refund{}
		if err = s.dao.LockApply(ctx, apply, req.BizNo); err != nil {
			return
		}
		switch pay.PayOrderStatus(apply.PayStatus) {
		case req.Status:
			return
		case pay.PayOrderStatus_REFUND_ORDER_STATUS_REFUNDING:
			// 只有退款申请中时接受推送
			if err = s.processRefundInTransaction(ctx, apply, req); err != nil {
				return
			}
		default:
			// 说明两边终态记录不一致，需要报警
			panic(fmt.Sprintf("退款通知时，售后状态与退款单状态不一致: apply_id = %d", apply.ID))
		}
		return
	}); err != nil {
		return
	}
	return
}

// 分摊结果
func (s *Service) PromotionResultHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &mall.PromotionOffApportionMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("promotion_notify: %s", string(msgBody)),
		msgBody, &err)
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		var details []*model.OrderDetail
		order := &model.Order{}
		if err = s.dao.LockOrder(ctx, order, req.OutOrderNo); err != nil {
			return
		}
		if order.PromotionStatus || order.Discount == 0 {
			// 冗余推送
			return
		}
		promotionMap := make(map[uint64]uint64)
		for _, detail := range req.Details {
			promotionMap[detail.OutOrderDetailId] = detail.Off
		}
		var promotionDetails []*mall.MallPromotionMessage_Detail
		if details, err = s.dao.FindOrderDetailByOrderIds(ctx, []uint64{order.ID}); err != nil {
			return
		}
		for _, detail := range details {
			if off, ok := promotionMap[detail.ID]; ok {
				detail.Discount = off
				if detail.Total >= detail.Discount {
					detail.Trans = detail.Total - detail.Discount
				} else {
					err = fmt.Errorf("分摊金额超过明细金额: detail_id=%d, total=%d, discount=%d",
						detail.ID, detail.Total, detail.Discount)
					return
				}
				if err = s.dao.Save(ctx, detail); err != nil {
					return
				}
				delete(promotionMap, detail.ID)
				promotionDetails = append(promotionDetails, &mall.MallPromotionMessage_Detail{
					StorageKey: fmt.Sprintf("%d-%d", detail.ProdId, detail.SkuId),
					Off:        off,
				})
			}
		}
		if len(promotionMap) != 0 {
			err = fmt.Errorf("促销推送订单明细与商城订单不匹配")
			return
		}
		order.PromotionStatus = true
		if err = s.dao.Save(ctx, order); err != nil {
			return
		}
		var promotionMessage json.RawMessage
		if promotionMessage, err = json.Marshal(&mall.MallPromotionMessage{
			OrderNumber: order.OrderNumber,
			Details:     promotionDetails,
			Timestamp:   req.Timestamp,
		}); err != nil {
			return
		}
		if err = s.dao.PublishUniqueNSQMessage(ctx, constant.TopicMallPromotion, promotionMessage); err != nil {
			return
		}
		return
	}); err != nil {
		return
	}
	return
}

// 仓库状态通知
func (s *Service) ProcessUpdateHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &inventory.InventoryOrderUpdatePayload{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("inventory_notify: %s", string(msgBody)),
		msgBody, &err)
	if req.Source != inventory.SourceEnum_MALL_ORDER {
		// 非商城履约单，丢弃
		return
	}
	var confirmSchedule, refundSchedule *model.Schedule
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		if req.DeliverMemo != nil {
			var inventoryRemarkJson json.RawMessage
			order := &model.Order{}
			if err = s.dao.LockOrder(ctx, order, req.ExtOrderNumber); err != nil {
				return
			}
			inventoryRemark := &model.InventoryRemark{}
			if err = json.Unmarshal(order.InventoryRemark, inventoryRemark); err != nil {
				err = ecode.WithStack(err)
				return
			}
			if req.Timestamp <= inventoryRemark.SyncTime || order.InventoryStatus == uint8(req.Status) {
				// 推送消息过期 & 推送消息重复
				return
			}
			switch req.Status {
			case inventory.ProcessOrderStatus_ProcessOrderReceived, inventory.ProcessOrderStatus_ProcessOrderCancelled:
				// 确认收货 & 取消发货均由商城发起，不接受仓库消息通知
				return
			case inventory.ProcessOrderStatus_ProcessOrderWaitMentioned, inventory.ProcessOrderStatus_ProcessOrderShipped:
				// 已发货 & 待自提，需要创建确认收货定时任务
				var config *model.Config
				if config, err = s.dao.GetConfig(ctx, order.SgId); err != nil {
					return
				}
				order.AutoConfirmTime = sql.NullTime{
					Valid: true, Time: time.Now().AddDate(0, 0, int(config.AutoConfirmedDays))}
				confirmSchedule = &model.Schedule{
					ScheduleType: uint8(mall.ScheduleType_SCHEDULE_TYPE_AUTO_CONFIRM),
					OrderId:      order.ID,
					ExecuteTime:  order.AutoConfirmTime.Time,
				}
				if err = s.dao.Create(ctx, confirmSchedule); err != nil {
					return
				}
			}
			order.InventoryStatus = uint8(req.Status)
			inventoryRemark.InventoryOrderUpdatePayload_DeliverMemo = *req.DeliverMemo
			if inventoryRemarkJson, err = json.Marshal(inventoryRemark); err != nil {
				err = ecode.WithStack(err)
				return
			}
			order.InventoryRemark = orm.JSON(inventoryRemarkJson)
			if err = s.dao.Save(ctx, order); err != nil {
				return
			}
		} else if req.RefundMemo != nil {
			var warehouseRemarkJson, payRemarkJson json.RawMessage
			apply := &model.Refund{}
			if err = s.dao.LockApply(ctx, apply, req.ExtOrderNumber); err != nil {
				return
			}
			warehouseRemark := &model.WarehouseRemark{}
			if err = json.Unmarshal(apply.WarehouseInfo, warehouseRemark); err != nil {
				err = ecode.WithStack(err)
				return
			}
			if req.Timestamp <= warehouseRemark.SyncTime || apply.InventoryStatus == uint8(req.Status) {
				// 推送消息过期 & 推送消息重复
				return
			}
			switch req.Status {
			case inventory.ProcessOrderStatus_ProcessOrderRejectReturn:
				// 拒绝接受
				apply.PayStatus = uint8(pay.PayOrderStatus_REFUND_ORDER_STATUS_FAILED)
				if payRemarkJson, err = json.Marshal(&model.RefundPaymentRemark{
					PayRemark:    model.PayRemark{},
					FailedReason: req.RefundMemo.Reason,
				}); err != nil {
					err = ecode.WithStack(err)
					return
				}
				apply.PayRemark = orm.JSON(payRemarkJson)
			case inventory.ProcessOrderStatus_ProcessOrderReturned:
				// 同意接受
				apply.PayStatus = uint8(pay.PayOrderStatus_REFUND_ORDER_STATUS_INIT)
				refundSchedule = &model.Schedule{
					ScheduleType: uint8(mall.ScheduleType_SCHEDULE_TYPE_REFUND_CREATE),
					OrderId:      apply.ID,
					ExecuteTime:  time.Now().Add(time.Second * 15),
				}
				if err = s.dao.Create(ctx, refundSchedule); err != nil {
					return
				}
			default:
				// 其他状态不同步
				return
			}
			apply.Status = uint8(mall.MallRefundStatus_MALL_REFUND_FINISHED)
			apply.InventoryStatus = uint8(req.Status)
			warehouseRemark.InventoryOrderUpdatePayload_RefundMemo = *req.RefundMemo
			if warehouseRemarkJson, err = json.Marshal(warehouseRemark); err != nil {
				err = ecode.WithStack(err)
				return
			}
			apply.WarehouseInfo = orm.JSON(warehouseRemarkJson)
			if err = s.dao.Save(ctx, apply); err != nil {
				return
			}
		} else {
			err = fmt.Errorf("仓库单%s配送&退货信息均为空", req.OrderNumber)
		}
		return
	}); err != nil {
		return
	}
	if refundSchedule != nil {
		if err = s.createRefund(ctx, refundSchedule.OrderId); err != nil {
			return
		}
	}
	return
}

func (s *Service) saveOrUpdateError(ctx context.Context, logBody string, msgBody []byte, rawError *error) {
	now := time.Now()
	retryMessage, _ := ctx.Value(msgRetryKey{}).(*model.NsqMessage)
	topic, _ := ctx.Value(nsq.TopicContextKey).(string)
	if *rawError != nil {
		// 只要失败，打印日志，仅保留单号记录
		log.FromContext(ctx).Warnw("nsq",
			"status_code", "Fail-Unknown",
			"method", "consume",
			"path", topic,
			"body", logBody,
			"error", (*rawError).Error(),
			"stacktrace", fmt.Sprintf("postgres error: %+v", *rawError),
		)
		if retryMessage == nil {
			if err := s.dao.Create(ctx, &model.NsqMessage{
				Topic:   topic,
				MsgKey:  uuid.New().String(), // 这里防止订单更新消息落地失败，失败消息key置换为uuid
				Payload: msgBody,
				ExecuteTime: sql.NullTime{
					Time:  now.Add(time.Minute),
					Valid: true,
				},
			}); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorSaving",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
			// 终止外部报错，允许消息正常结束；依赖定时任务触发后续步骤，不触发企业微信
			*rawError = nil
		} else {
			retryMessage.Retry += 1
			retryMessage.ExecuteTime = sql.NullTime{
				Time:  now.Add(time.Minute * time.Duration(retryMessage.Retry*retryMessage.Retry)),
				Valid: true,
			}
			if err := s.dao.Save(ctx, retryMessage); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorUpdating",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
			if retryMessage.Retry > 2 {
				// 重试超过2次，触发企业微信报警
				wechat.AsyncSendMarkDown(
					ctx, wechat.FormatBugMessage(
						ctx, conf.Conf.Env, conf.Conf.ServiceName, topic,
						string(msgBody),
						fmt.Sprintf("postgres error: %+v", *rawError)))
			}
		}
	} else {
		// 成功时打印简短日志，保留单号记录
		log.FromContext(ctx).Infow("nsq",
			"status_code", "Success",
			"method", "consume",
			"path", topic,
			"body", logBody,
		)
		if retryMessage != nil {
			// 已发送成功，更新重试任务
			retryMessage.ExecuteTime = sql.NullTime{}
			retryMessage.Status = 1
			if err := s.dao.Save(ctx, retryMessage); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorUpdating",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
		}
	}
	return
}
